package com.gxlevi.batch_process.task

import com.gxlevi.batch_process.bean.OrderRecordWide
import com.gxlevi.batch_process.util.HBaseUtil
import org.apache.flink.api.scala._

case class MerchantCountMoney(
                               var merchantId: String,
                               var date: String,
                               var amount: Double,
                               var count: Long
                             )

object MerchantCountMoneyTask {
  def process(orderRecordDataSet: DataSet[OrderRecordWide]) = {
    val mapDataSet = orderRecordDataSet.flatMap {
      orderRecordWide => {
        List(
          MerchantCountMoney(orderRecordWide.merchantId, orderRecordWide.yearMonthDay, orderRecordWide.payAmount.toDouble, 1),
          MerchantCountMoney(orderRecordWide.merchantId, orderRecordWide.yearMonth, orderRecordWide.payAmount.toDouble, 1),
          MerchantCountMoney(orderRecordWide.merchantId, orderRecordWide.year, orderRecordWide.payAmount.toDouble, 1)
        )
      }
    }

    val groupDataSet = mapDataSet.groupBy {
      merchant => (merchant.merchantId + merchant.date)
    }

    val reduceDataSet = groupDataSet.reduce {
      (p1, p2) => {
        MerchantCountMoney(p1.merchantId, p1.date, p1.amount + p2.amount, p1.count + p2.count)
      }
    }

    reduceDataSet.collect().foreach {
      merchant => {
        val tableName = "analysis_merchant"
        val rowKey = merchant.merchantId + ":" + merchant.date
        val clfName = "info"

        val merchantIdColumn = "merchantId"
        val dateColumn = "date"
        val amountColumn = "amount"
        val countColumn = "count"

        HBaseUtil.putMapData(tableName, rowKey, clfName, Map(
          merchantIdColumn -> merchant.merchantId,
          dateColumn -> merchant.date,
          amountColumn -> merchant.amount,
          countColumn -> merchant.count
        ))
      }
    }
  }
}
